package com.tang.consumer.connection;

import com.tang.consumer.netty.NettyClient;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author Tang
 * @classname ConnectManage
 * @description [  ]
 * @date 2021/1/24 19:22
 */
@Slf4j
@Component
public class ConnectManage {

    @Resource
    public NettyClient nettyClient;

    private AtomicInteger roundRobin = new AtomicInteger(0);
    /**
     * 为了顺序负载
     */
    private CopyOnWriteArrayList<Channel> channels = new CopyOnWriteArrayList<>();

    /**
     * channel 的维护 map
     */
    private Map<SocketAddress, Channel> channelMap = new ConcurrentHashMap<>(8);

    /**
     * 选择一个 Server Channel
     * @return
     */
    public Channel chooseChannel() {
        if (channels.size() > 0) {
            int size = channels.size();
            int index = (roundRobin.getAndAdd(1) + size) % size;
            return channels.get(index);
        }
        return null;
    }

    /**
     * 更新连接地址
     * @param addressList
     */
    public void updateConnectServer(List<String> addressList) {
        //1.没有可用的节点
        if (addressList == null || addressList.size() == 0) {
            log.error("没有可用的服务节点，全部服务器节点都不在线!");
            // 关闭所有的
            for (Channel channel : channels) {
                SocketAddress socketAddress = channel.remoteAddress();
                Channel node = channelMap.get(socketAddress);
                node.close();
            }
        }
        //2.存在可用的节点
        //存放最新节点下拉过来的注册地址
        HashSet<SocketAddress> newAllServerNode = new HashSet<>();
        addressList.forEach((address) -> {
            String[] split = address.split(":");
            newAllServerNode.add(new InetSocketAddress(split[0], Integer.parseInt(split[1])));
        });
        // 注册新的节点
        newAllServerNode.forEach((node) -> {
            // 连接新节点
            if (channelMap.get(node) == null) {
                connectServerNode(node);
            }
        });
        //去除无效的节点
        ArrayList<Integer> removeIndex = new ArrayList<>();
        for (int i = 0; i < channels.size(); i++) {
            Channel channel = channels.get(i);
            SocketAddress socketAddress = channel.remoteAddress();
            if (!newAllServerNode.contains(socketAddress)) {
                channelMap.remove(socketAddress).close();
                removeIndex.add(i);
                log.info("删除失效的服务节点:{}",socketAddress);
            }
        }
        removeIndex.forEach(channels::remove);
    }

    /**
     * 连接node
     * @param node
     * @return
     */
    private void connectServerNode(SocketAddress node) {
        Channel channel = null;
        try {
            channel = nettyClient.doConnect(node);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        assert channel != null;
        addChannel(channel,node);
    }

    /**
     * 添加 Channel
     * @param channel
     * @param node
     */
    private void addChannel(Channel channel,SocketAddress node) {
        log.info("加入Channel到连接管理器:{}",channel.remoteAddress());
        channels.add(channel);
        channelMap.put(node, channel);
    }

    /**
     * 删除 Channel
     * @param channel
     */
    public void remove(Channel channel) {
        log.info("移除Channel从连接管理器:{}",channel.remoteAddress());
        channels.remove(channel);
        channelMap.remove(channel.remoteAddress());
    }
}